package com.example.administrator.myapplication;

import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;
import android.view.View;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;

/**
 * observable  被观察者
 * observer    观察者
 * observable.subscribe(observer) 被观察者订阅观察者
 */
public class BaseUseActivity extends AppCompatActivity implements View.OnClickListener {

	@Override
	protected void onCreate(Bundle savedInstanceState) {
		super.onCreate(savedInstanceState);
		setContentView(R.layout.activity_base_use);

		findViewById(R.id.main_thread_rx).setOnClickListener(this);
		findViewById(R.id.io_thread_rx).setOnClickListener(this);

	}

	@Override
	public void onClick(View v) {
		switch (v.getId()) {
			case R.id.main_thread_rx:
				/*
				 *  RxJava 遵循的是线程不变的原则即：在哪个线程调用 subscribe()，
				 *  就在哪个线程生产事件；在哪个线程生产事件，就在哪个线程消费事件.
				 *
				 * */
				// 实例化被观察者
				final Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
					@Override
					public void subscribe(ObservableEmitter<String> emitter) throws Exception {
						Log.d("zwk", "被观察者:当前的线程" + Thread.currentThread());
						emitter.onNext("我是发送过来的数据");
					}
				});

				Observer<String> observer = new Observer<String>() {
					@Override
					public void onSubscribe(Disposable d) {

					}

					@Override
					public void onNext(String s) {
						Log.d("zwk", "当前运行的线程是" + Thread.currentThread().getName());
						Log.d("zwk", s);
					}

					@Override
					public void onError(Throwable e) {

					}

					@Override
					public void onComplete() {

					}
				};

				observable.subscribe(observer);
				break;

			/*
			 *Schedulers ——调度器，相当于线程控制器.
			 * RxJava 通过它来指定每一段代码应该运行在什么样的线程。
			 * RxJava 已经内置了几个 Scheduler
			 * subscribeOn()：指定，subscribe()发生的线程，事件产生的线程
			 * observeOn():事件消费的线程
			 * */
			case R.id.io_thread_rx:
				Observable<String> observable1 = Observable.create(new ObservableOnSubscribe<String>() {
					@Override
					public void subscribe(ObservableEmitter<String> emitter) throws Exception {
						Log.d("zwk", "observable的线程" + Thread.currentThread().getName());
						emitter.onNext("我是发送过来的信息");
					}
				})
						.subscribeOn(Schedulers.io())
						.observeOn(AndroidSchedulers.mainThread());


				Observer<String> observer1 = new Observer<String>() {
					@Override
					public void onSubscribe(Disposable d) {

					}

					@Override
					public void onNext(String s) {
						Log.d("zwk", "observer" + Thread.currentThread().getName());
					}

					@Override
					public void onError(Throwable e) {

					}

					@Override
					public void onComplete() {

					}
				};
				observable1.subscribe(observer1);
				break;

			default:
				break;
		}

	}
}
